Querying Qonto API and sending the results with Lambda
Previously, I wrote a blog about how to retrieve Qonto's bank transactions using its API and produce a CSV. The Python script was based on the local environment which can be manually executed at a specified period.
This time I will show you how I migrated it into AWS and automated the processings.
In this structure, I securely store the Qonto API keys in Secrets Manager. EventBridge triggers Lambda every Monday morning, which retrieves transaction data from Qonto with using the stored secrets, exports it to a CSV file, and email it as an attachment to me via SES.
Step 1 - Create a Lambda function
Let's create a new function by choosing "Author from scratch". For this script, I chose Python 3.8 as the Runtime. Other settings, including the execution role remain as default.
Step 2 - Configure other services
SES
Open "Email Addresses" on the SES console, click on [Verify a New Email Address], and fill in the email address(es) with which you want to send/receive the transaction data. The verification status is "pending verification" at this time.
Check the inbox of the email address that you specified, find the email sent from AWS, and click the link in order to verify the email address. Go back to the SES console and make sure that the verification status is now "verified".
Secrets Manager
Click on [Store a new secret] on the Secrets Manager console. Select "Other type of secrets", add your Qonto's login, secret key and IBAN under secret values and set the secret keys accordingly.
Once you've created the secret, you will see the sample code to retrieve the secret in your application, which you can use in the Lambda script.
EventBridge
Go back to the Lambda console, open the created function, and click on [Add trigger].
Select "EventBridge (CloudWatch Events)", choose "create a new rule", and fill in the rule name and description. As a schedule, choose "Schedule expression" and specify "cron(0 9 ? * MON *)" in order to trigger the function every Monday at 9 am (UTC). Uncheck "Enable trigger" for now and click [Add]. This is because the Lambda function is not yet ready to be run.
To learn more about cron expressions, see Schedule expressions using rate or cron.
Step 3 - Add IAM policies to the Lambda execution role
Create the following two policies on the IAM console and attach them to the Lambda execution role. For (1), please replace the resource ARN with your secret's ARN. You can find the Lambda role by going to Configuration > Permissions on the Lambda console.
(1) Allow Lambda to get the values of Secrets
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "secretsmanager:GetSecretValue",
"Resource": "arn:aws:secretsmanager:eu-west-1:123456789012:secret:Qonto-XXX"
}
]
}
(2) Allow Lambda to get send an email with SES
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"ses:SendEmail",
"ses:SendRawEmail"
],
"Resource": "*"
}
]
}
Step 4 - Upload modules and a script
First, prepare a directory where you collect the related modules. Then install the related modules (in this case, pytz and boto3) in the directory, place the python script ("lambda_function.py") as well, and then zip all these files. Do not zip the directory itself, because Lambda will then import the files under an extra hierarchy, which leads to the failure of loading files. You may also want to execute chmod in case the files do not have enough permissions to be read and executed.
mkdir lambda_files cd lambda_files pip install boto3 -t ./ pip install pytz -t ./ mv ../lambda_function.py . chmod -R 755 ./* zip -r zip_file ./*
Make sure that your local environment uses the Python version that corresponds to the Lambda runtime that you chose. (In this case, Python 3.8)
Ref: How to zip every module and upload a zip file to Lambda (Japanese only)
On the Lambda console, click [Upload from], select ".zip file" and choose the zip_file.zip.
Lambda script
Below are parts of the scripts. For the functions get_completes(), filter(), conv_utc(), conv_amount(), please see the previous blog, as I've only replaced print() with logger.log() and used the same processings.
# -*- coding:utf-8 -*-
import os
import boto3
import logging
import base64
import http.client, sys
from datetime import datetime, timedelta
import json, csv, pytz
import urllib.parse
from botocore.exceptions import ClientError
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
from email.mime.application import MIMEApplication
logger = logging.getLogger()
logger.setLevel(logging.INFO)
# Timezones
utc_tz = pytz.timezone('UTC')
local_tz = pytz.timezone('Europe/Berlin')
TRN, MEM = "transactions", "memberships"
AWS_REGION = "eu-west-1"
SENDER = 'aaa@example.com'
RECIPIENT = 'aaa@example.com'
SUBJECT = "Your weekly Qonto transactions"
CHARSET = "utf-8"
ses = boto3.client('ses', region_name=AWS_REGION)
def lambda_handler(event, context):
secret = get_secret() # Fetch secrets to use a Qonto API
dates, fdates = get_date() # Specify the period of filters for a Qonto request
# Get Pending/Declined transactions and if exists, display them in the log and email message
filters = filter(fdates, "update")
data = get_qonto(TRN, filters, secret)
updates_msg = log_updates(data)
# Get member data and Completed transactions
data_mem = get_qonto(MEM, filters, secret)
filters = filter(fdates, "settle")
data_set = get_qonto(TRN, filters, secret)
transactions = get_completes(data_mem, data_set)
if transactions:
try:
# Produce a csv file on Lambda
file_path = '/tmp/' + "qonto_" + dates[0] + "_" + dates[1] + ".csv"
with open(file_path, 'w', newline='') as csvFile:
csvwriter = csv.writer(csvFile, delimiter=',',quotechar='"', quoting=csv.QUOTE_NONNUMERIC)
csvwriter.writerow(['Buchungsdatum', 'Auftraggeber/Empfänger', 'Verwendungszweck', 'Betrag', 'Zusatzinfo'])
for record in transactions:
csvwriter.writerow(record)
except Exception as error:
logger.error(error)
logger.info("The csv file has been successfully generated: " + file_path)
body = "Hello,\r\n\r\nThe bank transactions of your Qonto accounts for the last week are now ready.\r\nPlease see the attached CSV."
else:
body = "Hello,\r\n\r\nThere are no bank transactions of your Qonto accounts for the last week.\r\n"
body = body + "\r\n\r\n" + '\r\n'.join(updates_msg)
# Send an email
send_raw_email(SENDER, RECIPIENT, SUBJECT, body, CHARSET, file_path)
def get_secret():
secret_name = "Qonto"
# Create a Secrets Manager client
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=AWS_REGION
)
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
raise e
else:
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
else:
secret = base64.b64decode(get_secret_value_response['SecretBinary'])
return json.loads(secret)
def get_date():
"""
Set a week ago at 0:00:00 at start date and yesterday at 23:59:59 as end date,
convert them from local timezone to UTC, and encode them.
Returns:
dates (list) : start/end datetime (YYYY-MM-DD)
filder_dates (list) : start/end datetime (YYYY-MM-DDThh:mm:ss.sss.Z)
"""
now = datetime.now(local_tz) # today in local timezone
start_datetime = now.replace(hour=0,minute=0,second=0,microsecond=0) + timedelta(days=-7)
end_datetime = now.replace(hour=23,minute=59,second=59,microsecond=999999) + timedelta(days=-1)
dates = []
filter_dates = []
for dt in start_datetime, end_datetime:
date = dt.strftime('%Y-%m-%d')
dates.append(date)
fdate = utc_tz.normalize(dt).strftime('%Y-%m-%dT%H:%M:%S.%fZ')
endate = urllib.parse.quote(fdate)
filter_dates.append(endate)
return dates, filter_dates
def log_updates(data):
num = data['meta']['total_count']
update_msg = []
msg = "There are {} non-settled records.".format(num)
update_msg.append(msg)
logger.info(msg)
for trn in data["transactions"]:
status = trn['status'] # status
date_utc = trn['updated_at'] # last updated date
amt = trn['amount'] # amount
side = trn['side'] # credit or debit
lab = trn['label'] # counterpart
ref = trn['reference'] # reference
# Convert date and amount
date = conv_utc(date_utc)
amount = conv_amount(side, amt)
# Log the results
msg = "[{} {}] Couterpart: {}, Amount: {} EUR, Reference: {}".format(status, date, lab, amount, ref)
logger.warning(msg)
update_msg.append(msg)
return update_msg
def get_qonto(typ, filters, secret):
"""
Get Qonto records with the specified parameters and (if applicable) filters.
Args:
typ (str): "transactions" or "memberships"
filters (str): filters of a request; applicable only for trn and consisted by status and dates
secret (array)
Returns:
output: json output of the fetched data
"""
seckey = secret['secret-key']
login = secret['login']
iban = secret['iban']
payload = "{}"
headers = { 'authorization': login + ":" + seckey}
param = typ # for transactions, addtl filters in the parameter
if typ == TRN: param = param + "?iban=" + iban + "&slug=" + login + filters
try:
conn = http.client.HTTPSConnection("thirdparty.qonto.com")
conn.request("GET", f"/v2/{param}", payload, headers)
except Exception as e:
logger.error(e)
logger.error("Error occurred while requesting {} to Qonto. Filters: {}".format(typ, filters))
raise e
res = conn.getresponse()
data = res.read()
output = json.loads(data.decode("utf-8"))
return output
def send_raw_email(src, to, sbj, body, char, file):
logger.info('send_raw_email: START')
msg = MIMEMultipart()
msg['Subject'] = sbj
msg['From'] = src
msg['To'] = to
msg_body = MIMEText(body.encode(char), 'plain', char)
msg.attach(msg_body)
att = MIMEApplication(open(file, 'rb').read())
att.add_header('Content-Disposition','attachment',filename=os.path.basename(file))
msg.attach(att)
try:
response = ses.send_raw_email(
Source=src,
Destinations=[to],
RawMessage={
'Data':msg.as_string()
}
)
except ClientError as e:
logger.error(e.response['Error']['Message'])
else:
logger.info("Email sent! Message ID:"),
logger.info(response['MessageId'])
Step 5 - Test and execute the Lambda
On the Lambda console, go to Test, specify an event name as you like, and click on [Invoke].
*The event is not used in this code, so you can leave it as default.
Hooray! I got the email!
You will see "succeeded" on the Lambda console. If the function failed, see the log and investigate the issue.
When the function is ready to be scheduled, don't forget to enable the EventBridge rule so that it automatically triggers the function for the future schedule.
